package com.lvw.hadoop.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.net.URI;

public class HDFSApp {

    private static final String HDFS_PATH = "hdfs://hadoop000:8020";
    FileSystem fileSystem = null;
    Configuration configuration = null;

    @Before
    public void setUp() throws Exception {
        System.out.println("HDFSApp.setUp");
        configuration = new Configuration();

        fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, "hadoop");
    }

    @After
    public void tearDown() throws Exception{
        configuration = null;
        fileSystem = null;
        System.out.println("HDFSApp.tearDown");
    }

    /**
     * make dir
     * @throws Exception
     */
    @Test
    public void mkdir() throws Exception {
        fileSystem.mkdirs(new Path("/hdfsapi/test"));
    }

    /**
     * create file
     *
     * @throws Exception
     */
    @Test
    public void create() throws  Exception {
        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/hdfsapi/test/hello"));
        fsDataOutputStream.write("Hello hadoop!".getBytes());
        fsDataOutputStream.flush();
        fsDataOutputStream.close();
    }

    /**
     * open file
     */
    @Test
    public void open() throws Exception {
        FSDataInputStream open = fileSystem.open(new Path("/hdfsapi/test/hello"));

        IOUtils.copyBytes(open, System.out, 1024);
    }

    /**
     * rename
     */
    @Test
    public void rename() throws Exception {
        Path oldPath = new Path("/hdfsapi/test/hello");
        Path newPath = new Path("/hdfsapi/test/hello.txt");
        boolean rename = fileSystem.rename(oldPath, newPath);
    }

    /**
     * upload
     */
    @Test
    public void copyFromLocal() throws  Exception {
        // D:\downloads\软件\client.exe
        Path local = new Path("D:\\downloads\\软件\\client.exe");
        Path hdfs = new Path("/hdfsapi/test");
        fileSystem.copyFromLocalFile(local, hdfs);
    }

    /**
     * upload with progress
     */
    @Test
    public void copyFromLocalWithProgress() throws  Exception {
        // D:\downloads\软件\Office_Project_Pro_2007_Win32_ChnSimp.zip
        Path path = new Path("/hdfsapi/test/Office_project.zip");
        BufferedInputStream inputStream = new BufferedInputStream(new FileInputStream("D:\\downloads\\软件\\Office_Project_Pro_2007_Win32_ChnSimp.zip"));
        FSDataOutputStream fsDataOutputStream = fileSystem.create(path, new Progressable() {
            @Override
            public void progress() {
                System.out.print(".");
            }
        });

        IOUtils.copyBytes(inputStream, fsDataOutputStream, 4096);

    }

    /**
     * download
     */
    @Test
    public void copyToLocal() throws Exception {
        Path hdfs = new Path("/hello.txt");
        Path local = new Path("D:\\temp");
        fileSystem.copyToLocalFile(false, hdfs, local, true);
    }

    /**
     * list file
     */
    @Test
    public void listFile() throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/hdfsapi/test/"));

        for (FileStatus status : fileStatuses) {
            String type = status.isDirectory() ? "Folder" : "File";
            long len = status.getLen();
            Path path = status.getPath();
            short replication = status.getReplication();
            System.out.println(type + "\t" + len + "\t" + path + "\t" + replication);
        }
    }

    /**
     * move file
     */
    @Test
    public void moveFile() throws Exception {
        Path fromPath = new Path("/hello.txt");
        Path toPath = new Path("/a/hello.txt");

        fileSystem.rename(fromPath, toPath);

    }

    /**
     * delete files recursive
     */
    @Test
    public void deleteFile() throws Exception {
        fileSystem.delete(new Path("/a"));
    }
}
